package com.kool.kadmin.mqtt.service;

import com.kool.kadmin.client.KauthClient;
import com.kool.kadmin.client.request.SetUserReq;
import com.kool.kadmin.client.request.SyncMqttUsersReq;
import com.kool.kadmin.client.request.SyncTopicPermitsReq;
import com.kool.kadmin.client.vo.MqttUserVo;
import com.kool.kadmin.client.vo.TopicPermitVo;
import com.kool.kadmin.common.bean.PageListResult;
import com.kool.kadmin.common.bean.Result;
import com.kool.kadmin.common.constant.ErrorCode;
import com.kool.kadmin.common.exception.AppException;
import com.kool.kadmin.common.util.PaginatorUtil;
import com.kool.kadmin.common.util.ResultUtil;
import com.kool.kadmin.common.util.SecurityUtil;
import com.kool.kadmin.mqtt.bean.MqttTopicAuthDo;
import com.kool.kadmin.mqtt.bean.MqttTopicDo;
import com.kool.kadmin.mqtt.bean.MqttUserDo;
import com.kool.kadmin.mqtt.constant.ActionEnum;
import com.kool.kadmin.mqtt.constant.MqttUserStateEnum;
import com.kool.kadmin.mqtt.dao.MqttTopicAuthDao;
import com.kool.kadmin.mqtt.dao.MqttTopicDao;
import com.kool.kadmin.mqtt.dao.MqttUserDao;
import com.kool.kadmin.mqtt.request.*;
import com.kool.kadmin.mqtt.vo.UserTopicPermitVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author : luyu
 * @date :2021/3/25 17:11
 */
@Service
@Slf4j
public class MqttAuthService {
    @Resource
    private MqttUserDao userDao;
    @Resource
    private MqttTopicAuthDao topicAuthDao;
    @Resource
    private KauthClient kauthClient;
    @Resource
    private MqttTopicDao topicDao;

    /**
     * 登记用户
     *
     * @param request
     */
    public void register(RegisterReq request) {
        String userName = request.getUserName();
        String pwd = request.getPwd();
        String pwdC = request.getPwdC();
        if (!pwd.equals(pwdC)) {
            throw new AppException(ErrorCode.BUSINESS_ERROR, "密码不一致");
        }
        //查询用户
        MqttUserDo userDo = userDao.getByUserName(userName);
        if (userDo != null) {
            throw new AppException(ErrorCode.BUSINESS_ERROR, "用户已存在");
        }
        //对密码做sha256加密
        String pwdSha = SecurityUtil.sha(pwd);

        userDo = new MqttUserDo();
        userDo.setUserName(userName)
                .setPwd(pwdSha)
                .setState(MqttUserStateEnum.ACTIVE.getCode());
        //入库
        userDao.insert(userDo);


        SetUserReq setUserReq = new SetUserReq();
        setUserReq.setUserName(userName);
        setUserReq.setPwd(pwdSha);
        setUserReq.setState(MqttUserStateEnum.ACTIVE.getCode());

        //往kauth发送注册请求
        kauthClient.setUser(setUserReq);
    }

    /**
     * 关闭用户
     *
     * @param request
     */
    public void closeUser(CloseUserReq request) {
        long id = request.getId();
        //查询用户
        MqttUserDo userDo = userDao.getById(id);
        if (userDo == null) {
            throw new AppException(ErrorCode.BUSINESS_ERROR, "用户不存在");
        }
        //检查状态
        if (!MqttUserStateEnum.ACTIVE.getCode().equals(userDo.getState())) {
            throw new AppException(ErrorCode.BUSINESS_ERROR,
                    String.format("用户状态为【%s】,只有激活状态的用户才能执行关闭操作", MqttUserStateEnum.getDescByCode(userDo.getState())));
        }

        userDo.setState(MqttUserStateEnum.CLOSED.getCode());
        //更新用户状态=关闭
        userDao.updateById(userDo);

        SetUserReq setUserReq = new SetUserReq();
        setUserReq.setUserName(userDo.getUserName());
        setUserReq.setPwd(userDo.getPwd());
        setUserReq.setState(MqttUserStateEnum.CLOSED.getCode());
        //往kauth发通知
        kauthClient.setUser(setUserReq);
    }

    /**
     * 删除用户
     *
     * @param request
     */
    @Transactional
    public void deleteUser(DeleteUserReq request) {
        long id = request.getId();
        //查询用户
        MqttUserDo userDo = userDao.getById(id);
        if (userDo == null) {
            throw new AppException(ErrorCode.BUSINESS_ERROR, "用户不存在");
        }
        //删除用户的主题权限
        topicAuthDao.deleteByUserName(userDo.getUserName());

        //删除用户
        userDao.deleteById(id);

        com.kool.kadmin.client.request.DeleteUserReq deleteUserReq = new com.kool.kadmin.client.request.DeleteUserReq();
        deleteUserReq.setUserName(userDo.getUserName());
        //往kauth发通知
        kauthClient.deleteUser(deleteUserReq);
    }

    /**
     * 激活用户
     *
     * @param request
     */
    public void openUser(OpenUserReq request) {
        long id = request.getId();
        //查询用户
        MqttUserDo userDo = userDao.getById(id);
        if (userDo == null) {
            throw new AppException(ErrorCode.BUSINESS_ERROR, "用户不存在");
        }
        //检查状态
        if (!MqttUserStateEnum.CLOSED.getCode().equals(userDo.getState())) {
            throw new AppException(ErrorCode.BUSINESS_ERROR,
                    String.format("用户状态为【%s】,只有关闭状态的用户才能执行激活操作", MqttUserStateEnum.getDescByCode(userDo.getState())));
        }

        userDo.setState(MqttUserStateEnum.ACTIVE.getCode());
        //更新用户状态=激活
        userDao.updateById(userDo);

        SetUserReq setUserReq = new SetUserReq();
        setUserReq.setUserName(userDo.getUserName());
        setUserReq.setPwd(userDo.getPwd());
        setUserReq.setState(MqttUserStateEnum.ACTIVE.getCode());
        //往kauth发通知
        kauthClient.setUser(setUserReq);
    }

    /**
     * 授权主题
     *
     * @param request
     */
    public void topicPermit(TopicPermitReq request) {
        String topicName = request.getTopicName();
        String userName = request.getUserName();
        Integer canSubscribe = request.getCanSubscribe();
        Integer canPublish = request.getCanPublish();
        //对权限做编码
        int authBits = (canSubscribe << ActionEnum.SUBSCRIBE.getBitIndex()) + (canPublish << ActionEnum.PUBLISH.getBitIndex());
        //查询主题授权用户信息是否存在
        MqttTopicAuthDo topicAuthDo = topicAuthDao.getByTopicUserName(topicName, userName);
        if (topicAuthDo == null) {
            //插入
            topicAuthDo = new MqttTopicAuthDo();
            topicAuthDo.setTopicName(topicName)
                    .setUserName(userName)
                    .setAuthBits(authBits);
            topicAuthDao.insert(topicAuthDo);
        } else {
            //更新
            topicAuthDo.setAuthBits(authBits);
            topicAuthDao.updateById(topicAuthDo);
        }

        com.kool.kadmin.client.request.TopicPermitReq topicPermitReq = new com.kool.kadmin.client.request.TopicPermitReq();
        topicPermitReq.setUserName(userName);
        topicPermitReq.setTopicName(topicName);
        topicPermitReq.setCanSubscribe(canSubscribe);
        topicPermitReq.setCanPublish(canPublish);

        //往kauth发送授权主题请求
        kauthClient.topicPermit(topicPermitReq);
    }

    /**
     * 分页查询用户
     *
     * @param request
     * @return
     */
    public PageListResult<MqttUserDo> getMqttUserPage(GetMqttUserPageReq request) {
        int currentPage = request.getCurrentPage();
        int pageSize = request.getPageSize();

        MqttUserDo condition = new MqttUserDo();
        condition.setState(request.getState());
        int total = userDao.countTotal(condition);
        List<MqttUserDo> list = userDao.getPage(condition, PaginatorUtil.calculateOffset(currentPage, pageSize), pageSize);

        return ResultUtil.success(list, currentPage, pageSize, total);
    }

    /**
     * 查询用户的主题权限
     *
     * @param request
     * @return
     */
    public PageListResult<UserTopicPermitVo> getUserTopicAuthPage(GetUserTopicAuthPageReq request) {
        String userName = request.getUserName();

        MqttTopicDo condition = new MqttTopicDo();
        //分页查询主题
        int total = topicDao.countTotal(condition);
        int offset = PaginatorUtil.calculateOffset(request.getCurrentPage(), request.getPageSize());
        List<MqttTopicDo> topicDos = topicDao.getPage(condition, offset, request.getPageSize());

        List<UserTopicPermitVo> list = new ArrayList<>();
        if (topicDos == null) {
            return ResultUtil.success(list, request.getCurrentPage(), request.getPageSize(), total);
        }
        for (MqttTopicDo topicDo : topicDos) {
            UserTopicPermitVo userTopicPermit = new UserTopicPermitVo();
            userTopicPermit.setTopicName(topicDo.getTopicName());
            //查询用户的主题权限
            MqttTopicAuthDo topicAuth = topicAuthDao.getByTopicUserName(topicDo.getTopicName(), userName);
            if (topicAuth != null) {
                int authBits = topicAuth.getAuthBits();
                //权限码解码
                userTopicPermit.setCanPublish((authBits >> ActionEnum.PUBLISH.getBitIndex()) & 0x01);
                userTopicPermit.setCanSubscribe((authBits >> ActionEnum.SUBSCRIBE.getBitIndex()) & 0x01);
            } else {
                userTopicPermit.setCanSubscribe(0);
                userTopicPermit.setCanPublish(0);
            }
            list.add(userTopicPermit);
        }
        return ResultUtil.success(list, request.getCurrentPage(), request.getPageSize(), total);
    }

    /**
     * 同步用户、主题权限数据到kauth
     */
    public void syncKauthRedis() {
        log.info("开始同步用户、主题权限数据到kauth");
        //同步用户逻辑
        int currentPage = 1;
        int pageSize = 100;
        //查询用户数
        int total = userDao.countTotal(new MqttUserDo());
        log.info("同步kauth: 用户数 = {}", total);
        if (total == 0) {
            return;
        }
        //分批处理用户
        while ((currentPage - 1) * pageSize < total) {
            List<MqttUserDo> users = userDao.getPage(new MqttUserDo(), PaginatorUtil.calculateOffset(currentPage, pageSize), pageSize);

            List<MqttUserVo> userVos = users.stream().map(item -> {
                MqttUserVo userVo = new MqttUserVo();
                userVo.setUserName(item.getUserName());
                userVo.setPwd(item.getPwd());
                userVo.setState(item.getState());
                return userVo;
            }).collect(Collectors.toList());

            SyncMqttUsersReq req = new SyncMqttUsersReq();
            req.setMqttUsers(userVos);
            //同步用户
            kauthClient.syncMqttUsers(req);

            log.info("同步kauth: 已同步用户数 = {}", (currentPage - 1) * pageSize + users.size());
            currentPage++;
        }

        //同步主题权限逻辑
        currentPage = 1;
        //查询主题权限数
        total = topicAuthDao.countTotal(new MqttTopicAuthDo());
        log.info("同步kauth: 主题权限数 = {}", total);
        if (total == 0) {
            return;
        }

        //分批处理主题权限
        while ((currentPage - 1) * pageSize < total) {
            List<MqttTopicAuthDo> topicAuthDos = topicAuthDao.getPage(new MqttTopicAuthDo(), PaginatorUtil.calculateOffset(currentPage, pageSize), pageSize);
            List<TopicPermitVo> topicPermitVos = topicAuthDos.stream().map(item -> {
                TopicPermitVo topicPermitVo = new TopicPermitVo();
                topicPermitVo.setUserName(item.getUserName());
                topicPermitVo.setTopicName(item.getTopicName());
                int authBits = item.getAuthBits();
                //权限码解码
                topicPermitVo.setCanPublish((authBits >> ActionEnum.PUBLISH.getBitIndex()) & 0x01);
                topicPermitVo.setCanSubscribe((authBits >> ActionEnum.SUBSCRIBE.getBitIndex()) & 0x01);
                return topicPermitVo;
            }).collect(Collectors.toList());

            SyncTopicPermitsReq syncTopicPermitsReq = new SyncTopicPermitsReq();
            syncTopicPermitsReq.setTopicPermits(topicPermitVos);
            //同步用户主题权限
            kauthClient.syncTopicPermits(syncTopicPermitsReq);
            log.info("同步kauth: 已同步主题权限数 = {}", (currentPage - 1) * pageSize + topicAuthDos.size());
            currentPage++;
        }

        log.info("结束同步用户、主题权限数据到kauth");
    }
}
